package mqtt

import (
	"errors"
	"fmt"
	"strconv"
	"sync"
	"time"

	"gitee.com/sansaniot/ssiot-core/facade/env"
	"gitee.com/sansaniot/ssiot-core/logger"
	mqtt "github.com/eclipse/paho.mqtt.golang"

	"ssforward/common/constant"
	"ssforward/common/utils"
)

type Client struct {
	Broker      string
	Port        string
	ClientId    string
	Username    string
	Password    string
	Timeout     time.Duration
	Handlers    map[string]mqtt.MessageHandler
	client      mqtt.Client
	connectLost sync.Map
}

func NewClient() *Client {
	c := &Client{
		ClientId: "ssforward-" + strconv.FormatInt(time.Now().UnixMilli(), 10),
		Broker:   env.DefaultConfig.Mqtt.Host,
		Port:     strconv.Itoa(int(env.DefaultConfig.Mqtt.Port)),
		Username: env.DefaultConfig.Mqtt.User,
		Password: env.DefaultConfig.Mqtt.Passwd,
		Timeout:  time.Duration(env.DefaultConfig.Mqtt.Timeout),
	}

	c.Handlers = make(map[string]mqtt.MessageHandler)
	return c
}

// 固定订阅(启动之前添加)
func (c *Client) InitSubscribe(topic string, h mqtt.MessageHandler) {
	c.Handlers[topic] = h
}

// 动态订阅
func (c *Client) Subscribe(topic string, h mqtt.MessageHandler) {
	c.Handlers[topic] = h
	c.client.Subscribe(topic, 0, h)
}

// 动态取消订阅
func (c *Client) UnSubscribe(topics ...string) {
	for _, t := range topics {
		delete(c.Handlers, t)
	}
	c.client.Unsubscribe(topics...)
}

func (c *Client) connectHandler(client mqtt.Client) {
	logger.Infof("Mqtt client %s connected", c.ClientId)
	value, _ := c.connectLost.LoadAndDelete("flag")
	if value != nil && value.(bool) {
		logger.Info("重连订阅主题")
		for t, h := range c.Handlers {
			client.Subscribe(t, 0, h)
			logger.Infof("Topic %s subscribed", t)
		}
	}
}

func (c *Client) connectLostHandler(client mqtt.Client, err error) {
	// to do reconnect
	fmt.Printf("Connect lost: %v", err)
}

func (c *Client) reconnectHandler(client mqtt.Client, ops *mqtt.ClientOptions) {
	c.connectLost.Store("flag", true)
	logger.Infof("Mqtt client %s reconnecting", c.ClientId)
}

func (c *Client) SetBroker(broker string) {
	c.Broker = broker
}

func (c *Client) SetPort(port string) {
	c.Port = port
}

func (c *Client) SetClientId(clientId string) {
	c.ClientId = clientId
}

func (c *Client) SetUsername(username string) {
	c.Username = username
}

func (c *Client) SetPassword(password string) {
	c.Password = password
}

func (c *Client) CreateClient() (mqtt.Client, error) {
	opts := mqtt.NewClientOptions()
	opts.AddBroker(fmt.Sprintf("tcp://%s:%s", c.Broker, c.Port))
	opts.SetClientID(c.ClientId)
	opts.SetUsername(c.Username)
	opts.SetPassword(c.Password)
	opts.SetConnectionLostHandler(c.connectLostHandler)
	opts.SetOnConnectHandler(c.connectHandler)
	opts.SetConnectTimeout(c.Timeout * time.Second)
	opts.SetCleanSession(true)
	opts.SetKeepAlive(60 * time.Second)
	opts.SetConnectRetryInterval(5 * time.Second)
	opts.SetAutoReconnect(true)
	opts.SetReconnectingHandler(c.reconnectHandler)
	c.client = mqtt.NewClient(opts)
	token := c.client.Connect()
	if token.Wait() && token.Error() == nil {
		c.sub(c.client)
	}
	return c.client, token.Error()
}

func (c *Client) sub(client mqtt.Client) {
	for t, h := range c.Handlers {
		client.Subscribe(t, 0, h)
		logger.Infof("Topic %s subscribed", t)
	}
}

func (c *Client) Publish(topic string, payload interface{}) error {
	if err := c.client.Publish(topic, 0, false, payload).Error(); err != nil {
		logger.Errorf("[%s]消息发布失败,err=%s", topic, err.Error())
		return err
	}
	return nil
}

func (c *Client) DisConnect() bool {
	//判断是否连接
	if c.client == nil || !c.client.IsConnected() {
		return true
	} else {
		//断开
		c.client.Disconnect(5)
		return true
	}
}

func (c *Client) IsConnected() bool {
	return c.client != nil && c.client.IsConnected()
}

// 同步下发任务id
var CidQueue sync.Map

// 同步下发
// topic 下发主题
// contentBytes 下发内容
// cid 命令id
// subT 同步订阅响应的主题
// fn 响应成功的回调函数
func (c *Client) SyncSend(topic string, contentBytes []byte, cid int, subT string, fn func(map[string]interface{}, *error)) error {
	var respErr error
	syncSendResp(c, subT, cid, fn, &respErr)
	_ = c.Publish(topic, contentBytes)
	CidQueue.LoadOrStore(cid, "")
	quit := make(chan string)
	t := time.NewTicker(1 * time.Second)
	over := time.NewTicker(5 * time.Second)
	defer over.Stop()
	defer t.Stop()
	go func() {
		for {
			select {
			case <-t.C:
				desc, ok := CidQueue.Load(cid)
				if !ok {
					// 响应成功
					quit <- ""
					return
				} else if len(desc.(string)) > 0 {
					// 响应失败
					quit <- desc.(string)
					return
				}
			case <-over.C:
				quit <- "下发超时"
				return
			}
		}
	}()
	errInfo := <-quit
	CidQueue.Delete(cid)
	if respErr != nil {
		errInfo = respErr.Error()
	}
	if len(errInfo) == 0 {
		return nil
	} else {
		return errors.New(errInfo)
	}
}

// 同步下发
func (c *Client) SyncSendTimer(topic string, contentBytes []byte, cid int, timeout time.Duration, subT string, fn func(map[string]interface{}, *error)) error {
	var respErr error
	syncSendResp(c, subT, cid, fn, &respErr)
	_ = c.Publish(topic, contentBytes)
	cidStr := strconv.Itoa(cid)
	constant.LocalCache.Set(cidStr, "", timeout)
	quit := make(chan string)
	t := time.NewTicker(1 * time.Second)
	over := time.NewTicker(timeout)
	defer over.Stop()
	defer t.Stop()
	go func() {
		for {
			select {
			case <-t.C:
				desc, exist := constant.LocalCache.Get(cidStr)
				if !exist {
					// 响应成功
					quit <- ""
					return
				} else if len(desc.(string)) > 0 {
					// 响应失败
					quit <- desc.(string)
					return
				}
			case <-over.C:
				quit <- "下发超时"
				return
			}
		}
	}()
	errInfo := <-quit
	constant.LocalCache.Delete(cidStr)
	if respErr != nil {
		errInfo = respErr.Error()
	}
	if len(errInfo) == 0 {
		return nil
	} else {
		return errors.New(errInfo)
	}
}

// 同步响应处理
// err 整个流程使用
func syncSendResp(c *Client, subT string, sid int, fn func(map[string]interface{}, *error), respErr *error) {
	c.Subscribe(subT, func(mqClient mqtt.Client, message mqtt.Message) {
		logger.Infof("订阅到主题%s, 消息%s", message.Topic(), string(message.Payload()))
		msgInfo := utils.BytesToMap(message.Payload())
		cf, ok := msgInfo["sid"]
		if !ok || cf == nil {
			return
		}
		respCid := int(cf.(float64))
		if respCid != sid {
			return
		}
		c.UnSubscribe(subT)
		cd, ok := msgInfo["code"]
		if !ok || cd == nil || cd.(float64) != 200 {
			*respErr = errors.New("终端命令响应异常")
		} else {
			// 响应成功,回调函数处理
			fn(msgInfo, respErr)
		}
		CidQueue.Range(func(k, v any) bool {
			if k.(int) == respCid {
				CidQueue.Delete(k)
				return false
			}
			return true
		})
	})
}
